package info.manxi;

import akka.actor.AbstractActor;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.PoisonPill;
import akka.actor.Props;
import akka.actor.Terminated;
import akka.actor.UntypedActor;
import akka.routing.ActorRefRoutee;
import akka.routing.RoundRobinRoutingLogic;
import akka.routing.Routee;
import akka.routing.Router;
import com.typesafe.config.ConfigFactory;

import java.util.ArrayList;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicBoolean;

public class RouterTest extends AbstractActor {

    public Router router;

    {
        ArrayList<Routee> routees = new ArrayList<>();
        for(int i = 0; i < 5; i ++) {
            //借用上面的 inboxActor
            ActorRef worker = getContext().actorOf(Props.create(InboxTest.class), "worker_" + i);
            getContext().watch(worker);//监听
            routees.add(new ActorRefRoutee(worker));
        }
        /**
         * RoundRobinRoutingLogic: 轮询
         * BroadcastRoutingLogic: 广播
         * RandomRoutingLogic: 随机
         * SmallestMailboxRoutingLogic: 空闲
         */
        router = new Router(new RoundRobinRoutingLogic(), routees);
    }


    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(InboxTest.Msg.class, o -> {
                    router.route(o, getSender());//进行路由转发
                })
                .match(Terminated.class, o -> {
                    router = router.removeRoutee(o.actor());//发生中断，将该actor删除。当然这里可以参考之前的actor重启策略，进行优化，为了简单，这里仅进行删除处理
                    System.out.println(o.actor().path() + " 该actor已经删除。router.size=" + router.routees().size());

                    if(router.routees().size() == 0){//没有可用actor了
                        System.out.print("没有可用actor了，系统关闭。");
                        flag.compareAndSet(true, false);
                        getContext().system().terminate();
                    }
                })
                .build();
    }


    public  static AtomicBoolean flag = new AtomicBoolean(true);

    public static void main(String[] args) throws InterruptedException {
        ActorSystem system = ActorSystem.create("router");
        ActorRef routerTest = system.actorOf(Props.create(RouterTest.class), "RouterTest");

        int i = 1;
        while(flag.get()){
            routerTest.tell(InboxTest.Msg.WORKING, ActorRef.noSender());

            if(i % 10 == 0) routerTest.tell(InboxTest.Msg.CLOSE, ActorRef.noSender());

            Thread.sleep(200);

            i ++;
        }
    }

    private static class InboxTest extends AbstractActor {
        @Override
        public Receive createReceive() {
            return receiveBuilder()
                    .matchAny(s -> {
                        System.err.println(getSelf() + "  == "+s);
                        if (Objects.equals(s, Msg.CLOSE)) getSelf().tell(PoisonPill.getInstance(),getSelf());
                    })
                    .build();
        }

        enum Msg {
             CLOSE, WORKING
         }
    }
}